use std::{
    future::{ready, Ready},
    mem::MaybeUninit,
    ops::Deref,
    sync::{atomic::{AtomicU64, Ordering}, Arc},
    time::Duration,
};

use actix_web::{dev::Payload, web::Bytes, FromRequest, HttpRequest};
use dashmap::DashMap;
use tokio::{
    sync::{mpsc, Mutex},
    time::timeout,
};

use crate::{
    error::{new_err, new_err_info, AppErr, ErrWrap, EC_DEVICE_NOT_LOGIN, EC_NO_EVENT, EC_NO_ID}, utils::current_timestamp, web::cfg::HTTP_DEVICE_SESSION_HEAD
};

use super::event::{
    get_event_manager, Event, CMD_NOTIFY_PRODUCT, EVENT_TYPE_ACK, EVENT_TYPE_NOTIFY, EVENT_TYPE_NOTIFY_ACK, EVENT_TYPE_REQ, EVENT_TYPE_RES
};

static mut SESSION_MAP: MaybeUninit<DashMap<i64, SharedDeviceState>> = MaybeUninit::uninit();

pub fn init() {
    unsafe {
        SESSION_MAP.write(DashMap::new());
    }
}

fn get_map() -> &'static DashMap<i64, SharedDeviceState> {
    unsafe { SESSION_MAP.assume_init_ref() }
}

#[derive(Debug)]
struct EventRx {
    rx: mpsc::Receiver<Event>,
    last: Option<Event>,
}

impl EventRx {
    async fn recv(&mut self) -> Result<Event, AppErr> {
        match self.last.as_ref() {
            Some(v) => return Ok(v.clone()),
            _ => {}
        };
        let event = self.rx.recv().await.ok_or(new_err("recv none"))?;
        self.last = Some(event.clone());
        Ok(event)
    }

    fn clear(&mut self) {
        self.last = None;
    }
}

#[derive(Debug)]
pub struct DeviceSession {
    pub id: i64,
    pub poll_timestamp: AtomicU64,
    event_rx: Mutex<EventRx>,
    event_tx: mpsc::Sender<Event>,
}

#[derive(Debug, Clone)]
pub struct SharedDeviceState(Arc<DeviceSession>);

impl Deref for SharedDeviceState {
    type Target = DeviceSession;

    fn deref(&self) -> &Self::Target {
        &self.0
    }
}

impl SharedDeviceState {

    pub fn new(id: i64) -> SharedDeviceState {
        let (tx, rx) = mpsc::channel(32);
        SharedDeviceState(Arc::new(DeviceSession {
            id,
            poll_timestamp: AtomicU64::new(current_timestamp()),
            event_rx: Mutex::new(EventRx { rx, last: None }),
            event_tx: tx,
        }))
    }

    pub async fn clear_last_event(&self) -> Result<(), AppErr> {
        let mut rx = timeout(Duration::from_secs(2), self.event_rx.lock())
            .await
            .wrap_msg("rx timeout")?;
        rx.clear();

        Ok(())
    }

    pub fn last_poll_timestamp(&self) -> u64 {
        self.poll_timestamp.load(Ordering::Relaxed)
    }

    pub async fn recv_event(&self) -> Result<Event, AppErr> {
        self.poll_timestamp.store(current_timestamp(), Ordering::Release);

        let mut rx = timeout(Duration::from_secs(2), self.event_rx.lock())
            .await
            .wrap_msg("rx timeout")?;

        let event = timeout(Duration::from_secs(8), rx.recv())
            .await
            .map_err(|_| new_err_info(EC_NO_EVENT, "没有新的事件"))??;

        Ok(event)
    }

    pub async fn notify(&self, cmd: u8, body: Bytes) -> Result<(), AppErr> {
        let event = Event {
            ctype: EVENT_TYPE_NOTIFY,
            seq: None,
            cmd: Some(cmd),
            ec: None,
            body,
        };

        self.event_tx.send(event).await.wrap()?;

        Ok(())
    }

    pub async fn notify_ack(&self, cmd: u8, body: Bytes) -> Result<(), AppErr> {
        let manager = get_event_manager();
        let (rx, mut destroy) = manager.create_rx(EVENT_TYPE_ACK);

        let event = Event {
            ctype: EVENT_TYPE_NOTIFY_ACK,
            seq: Some(destroy.seq),
            cmd: Some(cmd),
            ec: None,
            body,
        };

        self.event_tx.send(event).await.wrap()?;

        timeout(Duration::from_secs(10), rx)
            .await
            .wrap_msg("recv timeout")?
            .wrap()?;

        destroy.ok();
        Ok(())
    }

    pub async fn req(&self, cmd: u8, body: Bytes) -> Result<Event, AppErr> {
        let manager = get_event_manager();
        let (rx, mut destroy) = manager.create_rx(EVENT_TYPE_RES);

        let event = Event {
            ctype: EVENT_TYPE_REQ,
            seq: Some(destroy.seq),
            cmd: Some(cmd),
            ec: None,
            body,
        };

        self.event_tx.send(event).await.wrap()?;

        let event = timeout(Duration::from_secs(30), rx)
            .await
            .wrap_msg("recv timeout")?
            .wrap()?;

        destroy.ok();

        Ok(event)
    }
}

pub fn save_session(id: i64) {
    let map = get_map();
    map.insert(id, SharedDeviceState::new(id));
}

pub fn remove_session(id: i64) {
    let map = get_map();
    map.remove(&id);
}

pub fn get_device_session(id: i64) -> Result<SharedDeviceState, AppErr> {
    let map = get_map();
    let state = map
        .get(&id)
        .ok_or(new_err_info(EC_DEVICE_NOT_LOGIN, "该设备没有登陆"))?
        .value()
        .clone();
    Ok(state)
}

pub async fn notify_product_update() -> Result<(), AppErr> {
    let map = get_map();
    for d in map.iter() {
        d.value().notify(CMD_NOTIFY_PRODUCT, Bytes::new()).await?;
    }
    Ok(())
}

fn create_session(req: &HttpRequest) -> Result<SharedDeviceState, AppErr> {
    let id: i64 = req
        .headers()
        .get(HTTP_DEVICE_SESSION_HEAD)
        .ok_or(new_err("not device token"))?
        .to_str()
        .wrap()?
        .parse()
        .wrap()?;

    let state = get_map()
    .get(&id)
    .ok_or(new_err_info(EC_NO_ID, "没有id信息"))?;

    Ok(state.value().clone())
}

impl FromRequest for SharedDeviceState {
    type Error = AppErr;

    type Future = Ready<Result<Self, AppErr>>;

    fn from_request(req: &HttpRequest, _payload: &mut Payload) -> Self::Future {
        ready(create_session(req))
    }
}
